package com.atguigu.pj.app

import com.atguigu.cm.util.StrUtil
import com.atguigu.pj.acc.AccUserAction
import com.atguigu.pj.bean.{CategoryCountInfo, UserVisitAction}
import com.atguigu.pj.cons.ConsValue
import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD

import scala.collection.mutable

/**
 * description ：Top10 热门品类  按照每个品类的 点击、下单、支付 的量来统计热门品类（按点击，下单，支付数量来排序，如果相同，就按下一个指标排序）
 * author      ：剧情再美终是戏
 * mail        : 13286520398@163.com
 * date        ：Created in 2020/1/10 7:46
 * modified By ：
 * version:    : 1.0
 */
object QueryCategoryTop10 {


  /**
   * Top10 热门品类
   *
   * @Author 剧情再美终是戏
   * @Date 2020/1/10 11:49
   * @param sc   sparkContext
   * @param list 源数据
   * @return scala.collection.Seq<com.atguigu.pj.bean.CategoryCountInfo>
   * @Version 1.0
   **/
  def doQuery(sc: SparkContext, list: RDD[UserVisitAction]): List[CategoryCountInfo] = {

    // 初始化累加载器并向sparkcontext注册
    val accUserAction = new AccUserAction
    sc.register(accUserAction)

    // 遍历源数据，通过累加载器计算点击、下单、支付数量
    list.foreach({
      case visitAction: UserVisitAction =>
        // TODO 点击、下单、支付数量的优先级别：支付数量-->下单-->点击
        // 1、如果支付的品类id不等于null,支付的商品id不等于null
        if (StrUtil.isNotEmpty(visitAction.pay_category_ids) && StrUtil.isNotEmpty(visitAction.pay_product_ids)) {
          for (id <- visitAction.pay_category_ids.split(",")) accUserAction.add(id, ConsValue.user_action_pay)
        }
        // 2、如果下单的品类id不等于null,支付的商品id不等于null
        else if (StrUtil.isNotEmpty(visitAction.order_category_ids) && StrUtil.isNotEmpty(visitAction.order_product_ids)) {
          for (id <- visitAction.order_category_ids.split(",")) accUserAction.add(id, ConsValue.user_action_order)
        }
        // 3、点击品类id,点击商品id不等于-1
        else if (visitAction.click_category_id != (-1) && visitAction.click_product_id != (-1)) {
          accUserAction.add(visitAction.click_category_id.toString, ConsValue.user_action_click)
        }
    })

    // 获取累加器的值
    val accValue: mutable.Map[(String, String), Long] = accUserAction.value


    // 对分类id进行分组
    val actionCountByCategoryMap: Map[String, mutable.Map[(String, String), Long]] = accValue.groupBy(_._1._1)

    // 对分组好的记录封装到 CategoryCountInfo 类中
    val infoes: List[CategoryCountInfo] = actionCountByCategoryMap.map {
      case (cid, vMap) =>
        CategoryCountInfo(
          cid,
          vMap.getOrElse((cid, ConsValue.user_action_click), 0L),
          vMap.getOrElse((cid, ConsValue.user_action_order), 0L),
          vMap.getOrElse((cid, ConsValue.user_action_pay), 0L)
        )
    }.toList

    // 对Iterable[CategoryCountInfo]数据时行排序，取top10
    val result = infoes.sortBy(info =>
      (info.clickCount, info.orderCount, info.payCount))(Ordering.Tuple3(Ordering.Long.reverse, Ordering.Long.reverse, Ordering.Long.reverse)).take(2)

    // 返回结果
    result
  }
}
